#include <sys/types.h>
#include <sys/socket.h>
#include <sys/uio.h>
#include <stdio.h>
#include <netinet/in.h>
#include <sys/time.h>
#include <sys/ioctl.h>
#include <sys/stat.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#include <fcntl.h>

#include "libmod.h"
#include "log.h"
#include "confparser.h"

#include "mc_gossip.h"

int mod_init(char *);

static ssize_t mc_gossip_handler(WTI *wtip, CONN_INFO *cip, void *data, void **res);
static ssize_t mc_gossip_init(void *arg, void **res);

int process_cmd_reg(MC_INFO *gossip_info, int sendback_fd);
int process_cmd_disreg(MC_INFO *gossip_info, int sendback_fd);
int process_cmd_gspset(MC_INFO *gossip_info, int sendback_fd);
int process_cmd_set(MC_INFO *gossip_info, int sendback_fd);

static void msgid_list_init(void);
static void msgid_list_add(uint64_t msgid);
static int msgid_is_exist(uint64_t msgid);


static GOSSIP_HANDLER *gsp_handler;

/* config buffer (sl.conf) */
static char init_host[CONF_ITEM_LEN + 1];
static char current_host[CONF_ITEM_LEN + 1];
static int init_port;
static int current_port;
static int default_ttl;
static int peer_id;

static char _mod_name[CONF_ITEM_LEN + 1];

static struct conf_int_config conf_int_array[] = {
	{"init_port", &init_port},
	{"current_port", &current_port},
	{"default_ttl", &default_ttl},
	{"peer_id", &peer_id},
	{0, 0}
};

static struct conf_str_config conf_str_array[] = {
	{"init_host", init_host},
	{"current_host", current_host},
	{0, 0}
};


/* a queue to save the last MSG_HISTORY_LIST_LEN message's id */
static struct _msgids
{
    uint64_t list[MSG_HISTORY_LIST_LEN];
    int index;
    int size;
    pthread_rwlock_t    list_lock;
}msgids;


int 
process_cmd_reg(MC_INFO *gossip_info, int sendback_fd)
{
    char msg_buf[MAX_MSG_LEN + 1];
    int fd;

    FILE *file;
    uint64_t rand_num;

    file = fopen("/dev/random", "r");
    fread(&rand_num, sizeof(uint64_t), 1, file);
    fclose(file);
    
    #ifdef DEBUG
    printf("(msg)[%d,%s:%d] received PEER message from [%d]:\n", gsp_handler->localpeer->l_peerid, gsp_handler->localpeer->l_host, 
                                                gsp_handler->localpeer->l_port, gossip_info->sender_id);
    printf("\tgsp_reg %s %d %d %d %lu %d %d\n", 
                gossip_info->host, gossip_info->port, gossip_info->peer_id,
                gossip_info->ttl, gossip_info->msg_id, gossip_info->sender_id, gossip_info->src_id);
    #endif

    gossip_writen(sendback_fd, "DONE\r\n", 6);
    //write(sendback_fd, "DONE\r\n", 6);

    /* build the message then gossip the message if TTL > 0 */
    gossip_info->ttl--;

    /* if command is received just now, do not excute, but gossip */
    if (msgid_is_exist(gossip_info->msg_id) == 0)
    {
        /* add the peer to peer list */
        if (gossip_addpeer(gsp_handler, gossip_info->host, gossip_info->port, 
                        gossip_info->peer_id) != GSP_PEER_ADD_OK)
        {

            #ifdef DEBUG
            printf("(warn) add peer returns error! peer id may repeated.\n");
            #endif
            
            WLOG_ERR("(warn) add peer returns error! peer id may repeated.\n");
        } 

        if (gossip_info->ttl > 0)
        {
            /* 
             * register myself to the message src.
             * PEER A register to PEER B then
             * PEER B also register to PEER A.
             */
         
            fd = gossip_connect(gossip_info->host, gossip_info->port);

            if (fd >= 0)
            {
                /* XXX notice that here TTL = 1 */
          
                snprintf(msg_buf, MAX_MSG_LEN, "gsp_reg %s %d %d %d %lu %d\r\n", 
                    gsp_handler->localpeer->l_host, gsp_handler->localpeer->l_port, 
                    gsp_handler->localpeer->l_peerid, 1, rand_num, gsp_handler->localpeer->l_peerid);  

                #ifdef DEBUG
                printf("(msg) register to: [%d,%s:%d]\n", gossip_info->peer_id, gossip_info->host, gossip_info->port);
                printf("\t%s\n", msg_buf);
                #endif

                gossip_writen(fd, msg_buf, strlen(msg_buf));
            }
        }
    }
    
    
    if (gossip_info->ttl > 0)
    { 
        /* starts to gossip */
        snprintf(msg_buf, MAX_MSG_LEN, "gsp_reg %s %d %d %d %lu %d %d\r\n", 
                gossip_info->host, gossip_info->port, gossip_info->peer_id,
                gossip_info->ttl, gossip_info->msg_id, gsp_handler->localpeer->l_peerid, gossip_info->src_id);  
          
        if (gossip_route(gsp_handler, msg_buf, strlen(msg_buf), gossip_info->msg_id, 
                        gossip_info->src_id, gossip_info->sender_id) != GSP_ROUTE_OK)
        {

            #ifdef DEBUG
            printf("(warn) gossip_route. no message routed.\n");
            #endif
            
            WLOG_INFO("(warn) gossip_route. no message routed.\n");
        }
    }
    else
    {
        #ifdef DEBUG
        printf("(warn) TTL <= 0 nothing done.\n");
        #endif
        do{}while(0);
    }

    return 0;
}

int 
process_cmd_disreg(MC_INFO *gossip_info, int sendback_fd)
{
    char msg_buf[MAX_MSG_LEN + 1];
    
    #ifdef DEBUG
    printf("(msg)[%d,%s:%d] received PEER message from [%d]:\n", gsp_handler->localpeer->l_peerid, gsp_handler->localpeer->l_host, 
                    gsp_handler->localpeer->l_port, gossip_info->sender_id);

    printf("\tgsp_disreg %d %d %lu %d %d\n", 
                gossip_info->peer_id,gossip_info->ttl, gossip_info->msg_id, gossip_info->sender_id, gossip_info->src_id);
    #endif

    gossip_writen(sendback_fd, "DONE\r\n", 6);
    //write(sendback_fd, "DONE\r\n", 6);
  
    
    if (msgid_is_exist(gossip_info->msg_id) == 0)
    {
        /* add the peer to peer list */
        if (gossip_rmpeer(gsp_handler, gossip_info->peer_id, 0) != GSP_PEER_REMOVE_OK)
        {
            #ifdef DEBUG
            printf("(warn) remove peer returns error! may be peer do not exist.\n");
            #endif

            WLOG_ERR("(warn) remove peer returns error! may be peer do not exist.\n");
        }
        else
        {
            #ifdef DEBUG
            printf("(msg) remove peer returns OK!\n");
            #endif

            WLOG_INFO("(msg) remove peer returns OK!\n");
        }
    }
    
    /* build the message then gossip the message if TTL > 0 */
    gossip_info->ttl--;

    if (gossip_info->ttl > 0)
    {
        snprintf(msg_buf, MAX_MSG_LEN, "gsp_disreg %d %d %lu %d %d\r\n", 
                gossip_info->peer_id, gossip_info->ttl, gossip_info->msg_id, gsp_handler->localpeer->l_peerid, gossip_info->src_id);  
          
        if (gossip_route(gsp_handler, msg_buf, strlen(msg_buf), gossip_info->msg_id, 
                        gossip_info->src_id, gossip_info->sender_id) != GSP_ROUTE_OK)
        {
            #ifdef DEBUG
            printf("(warn) gossip_route. no msg routed.\n");
            #endif
            
            WLOG_INFO("(warn) gossip_route. no msg routed.\n");
        }
    }
    else
    {
        #ifdef DEBUG
        printf("(warn) TTL <= 0 nothing done.\n");
        #endif
        do{}while(0);
    }

    return 0;
}

int 
process_cmd_gspset(MC_INFO *gossip_info, int sendback_fd)
{
    char msg_buf[MAX_MSG_LEN + 1];
    
    #ifdef DEBUG
    printf("(msg)[%d,%s:%d] received PEER message from [%d]:\n", gsp_handler->localpeer->l_peerid, gsp_handler->localpeer->l_host, 
                        gsp_handler->localpeer->l_port, gossip_info->sender_id);

    printf("\tgsp_set %s %s %s %lu %d %lu %d %d\n", 
                gossip_info->keys[0],gossip_info->flag, gossip_info->exptime,
                gossip_info->bytes, gossip_info->ttl, gossip_info->msg_id, gossip_info->sender_id, gossip_info->src_id);
    #endif

    gossip_writen(sendback_fd, "DONE\r\n", 6);
    //write(sendback_fd, "DONE\r\n", 6);
    
    /* build the message then gossip the message if TTL > 0 */
    gossip_info->ttl--;

    if (gossip_info->ttl > 0)
    {
        snprintf(msg_buf, MAX_MSG_LEN, "gsp_set %s %s %s %lu %d %lu %d %d\r\n", 
                 gossip_info->keys[0],gossip_info->flag, gossip_info->exptime,
                gossip_info->bytes, gossip_info->ttl, gossip_info->msg_id, gsp_handler->localpeer->l_peerid, gossip_info->src_id);  
          
        if (gossip_route(gsp_handler, msg_buf, strlen(msg_buf), gossip_info->msg_id, 
                        gossip_info->src_id, gossip_info->sender_id) != GSP_ROUTE_OK)
        {

            #ifdef DEBUG
            printf("(warn) gossip_route. no msg routed.\n");
            #endif
            
            WLOG_INFO("(warn) gossip_route. no msg routed.\n");
        }
    }
    else
    {
        #ifdef DEBUG
        printf("(warn) TTL <= 0 nothing done.\n");
        #endif
        do{}while(0);
    }

    return 0;
}

int
process_cmd_set(MC_INFO *gossip_info, int sendback_fd)
{
    char msg_buf[MAX_MSG_LEN + 1];

    FILE *file;
    uint64_t rand_num;

    file = fopen("/dev/random", "r");
    fread(&rand_num, sizeof(uint64_t), 1, file);
    fclose(file);


    #ifdef DEBUG
    printf("(msg)[%d,%s:%d] received CLIENT message:\n", gsp_handler->localpeer->l_peerid, gsp_handler->localpeer->l_host, 
                                                gsp_handler->localpeer->l_port);

    printf("\tset %s %s %s %lu\n", 
                gossip_info->keys[0],gossip_info->flag, gossip_info->exptime,
                gossip_info->bytes);
    #endif

    if (msgid_is_exist(gossip_info->msg_id) == 0)
    {
        /*
         * XXX DO THE THINGS THAT "SET" SHOULD DO
         */
        do{}while(0);
    }

    gossip_writen(sendback_fd, "STORED\r\n", 8);
    //write(sendback_fd, "STORED\r\n", 8);
    

    snprintf(msg_buf, MAX_MSG_LEN, "gsp_set %s %s %s %lu %d %lu %d %d\r\n", 
            gossip_info->keys[0],gossip_info->flag, gossip_info->exptime, gossip_info->bytes, 
            gsp_handler->localpeer->l_ttl, rand_num, gsp_handler->localpeer->l_peerid, gsp_handler->localpeer->l_peerid);  
          
    if (gossip_route(gsp_handler, msg_buf, strlen(msg_buf), rand_num, 
                    gsp_handler->localpeer->l_peerid, 0) != GSP_ROUTE_OK)
    {

        #ifdef DEBUG
        printf("error: gossip_route. no msg routed.\n");
        #endif
            
        WLOG_INFO("(warn) gossip_route. no msg routed.\n");
    }

    

    return 0;
}



static ssize_t 
mc_gossip_handler(WTI *wtip, CONN_INFO *cip, void *data, void **res)
{
    PROTOCOL *protocol;
    MC_INFO *gossip_info;
    int sendback_fd;
    int rest;

    CSF_UNUSED_ARG(res);
    CSF_UNUSED_ARG(wtip);
    CSF_UNUSED_ARG(cip);
	
	protocol = (PROTOCOL *)data;
    gossip_info = (MC_INFO *)get_user_data(protocol);

    /* get the fd via CSF struct CONN_INFO. thus 'protocol.h' is included */
    sendback_fd = ((CONN_INFO *)(get_custom_p2(protocol)))->fd;

    if (protocol == NULL || gossip_info == NULL)
    {

        #ifdef DEBUG
        printf("(err) can not get PROTOCOL handler or GOSSIP_INFO handler. failed.\n");
        #endif
        return (PIPELINE_FAILED);
    }

    switch (gossip_info->cmd)
    {
        case _cmd_reg : 
            rest = process_cmd_reg(gossip_info, sendback_fd);
            break;

        case _cmd_disreg :
            rest = process_cmd_disreg(gossip_info, sendback_fd);
            break;

        case _cmd_gspset :
            rest = process_cmd_gspset(gossip_info, sendback_fd);
            break;

        case _cmd_set :
            rest = process_cmd_set(gossip_info, sendback_fd);
            break;

        default :
            rest = -1;
    }

    /* 
     * gossip the removing peer list.
     * this function is forbidden now
     */
    //gossip_disreg(gsp_handler);

    /* add the message to message history list */
    msgid_list_add(gossip_info->msg_id);

    /* set the command back to _cmd_null */
    gossip_info->cmd = _cmd_null;

    if (rest == -1)
    {
        #ifdef DEBUG
        printf(">>>error or nothind done at gossip. \n");
        #endif
        //return (PIPELINE_CONTINUE);
    }

    #ifdef DEBUG  
    printf("\n#####################################################\n\n");
    #endif

    return (PIPELINE_CONTINUE);
}


static void
msgid_list_init()
{
    msgids.index = 0;
    msgids.size = MSG_HISTORY_LIST_LEN;
    memset(msgids.list, 0, sizeof(uint64_t) * MSG_HISTORY_LIST_LEN);
    pthread_rwlock_init(&msgids.list_lock, NULL);
}


static void
msgid_list_add(uint64_t msgid)
{
    pthread_rwlock_wrlock(&msgids.list_lock);

    if (msgids.index < msgids.size)
    {
        msgids.list[msgids.index] = msgid;
        msgids.index++;
    }
    else
    {
        msgids.index = 0;
        msgids.list[msgids.index] = msgid;
        msgids.index++;
    }
    pthread_rwlock_unlock(&msgids.list_lock);
}


static int 
msgid_is_exist(uint64_t msgid)
{
    int i;
    int res = 0;

    if (pthread_rwlock_rdlock(&msgids.list_lock) != 0)
    {
        res = 0;
    }
    else
    {
        for (i = 0; i < msgids.size; i++)
        {
            if (msgids.list[i] == msgid)
            {
                res = 1;
                break;
            }
        }
    }

    pthread_rwlock_unlock(&msgids.list_lock);

    return res;
}


static ssize_t
mc_gossip_init(void *arg, void **res)
{
	int r;

    CSF_UNUSED_ARG(arg);
    CSF_UNUSED_ARG(res);

    /* load string config */
	r = load_conf(NULL, _mod_name, NULL, conf_str_array);
    
    /* load integer config */
    r = load_conf(NULL, _mod_name, conf_int_array, NULL);
	
	
	if (r != 0)
	{
	    WLOG_ERR(" load_conf() failed!");
	}

    /* init message history list */
    msgid_list_init();

    /* init gossip */
    gsp_handler = gossip_init(init_host, init_port, current_host, 
						current_port, default_ttl, peer_id);
    
    
    return 0;
}


int 
mod_init(char *mod_name)
{
	strlcpy(_mod_name, mod_name, CONF_ITEM_LEN + 1);
	
	set_request_init(mc_gossip_init);
	set_request_deinit(NULL);
	set_request_handler(mc_gossip_handler);

	return (0);
}





